package com.cxz.kafka;

import org.apache.kafka.clients.producer.Callback;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Objects;

/**
 * kafka生产
 * @author ocxz1
 */
public class KafkaProducerTest {

    private static final Logger logger = LoggerFactory.getLogger(KafkaProducerTest.class);

    /**
     * 获取kafka生产端
     * @return 返回kafka生产端
     */
    public static KafkaProducer<String, String> getProducer() {

        // 根据默认Kafka生产者配置，创建
        return  new KafkaProducer<>(KafkaCommonProperties.getDefaultKafkaProducerConfig());
    }

    public static void main(String[] args) {

        // 获取生产端

        // topic
        try (KafkaProducer<String, String> producer = KafkaProducerTest.getProducer()) {
            String topic = KafkaCommonProperties.TOPIC.toArray()[0].toString();
            for (int i = 0; i < 10; i++) {
                // 异步写入数据
                ProducerRecord<String, String> producerRecord = new ProducerRecord<>(topic, "消息" + i);
                producer.send(producerRecord, new MyProducerCallback());

            }
        } catch (Exception e) {
            logger.error(e.getMessage());
        }
    }

    /**
     * 异步数据写入完成后回调
     */
    static class MyProducerCallback implements Callback {
        @Override
        public void onCompletion(RecordMetadata recordMetadata, Exception e) {
            if (Objects.nonNull(e)) {
                logger.error(e.getMessage());
            }

            if (Objects.nonNull(recordMetadata)) {
                logger.info(">>>>>>>>>>Producer生产消息：metadata：{}，partition:{}, offset：{}", recordMetadata, recordMetadata.partition(), recordMetadata.offset());
            }
        }
    }
}
